package com.pap.rabbitmq.sender;

import java.util.UUID;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.pap.rabbitmq.config.RabbitMQConfig;
import com.pap.rabbitmq.data.MyCorrelationData;

@Service
public class MsgSender implements ConfirmCallback, ReturnCallback, InitializingBean {

	static class Student {
		private String id;
		private String name;

        public Student() {
        }

		public String getId() {
			return id;
		}

		public void setId(String id) {
			this.id = id;
		}

		public String getName() {
			return name;
		}

		public void setName(String name) {
			this.name = name;
		}
	}

    @Bean(name = "myRabbitTemplate")
    public RabbitTemplate myRabbitTemplate(ConnectionFactory connectionFactory) {
        // TODO logger ("==> custom rabbitTemplate, connectionFactory:"+ connectionFactory);
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(jackson2JsonMessageConverter());
        return rabbitTemplate;
    }

    @Bean
    public Jackson2JsonMessageConverter jackson2JsonMessageConverter() {
        Jackson2JsonMessageConverter jsonMessageConverter = new Jackson2JsonMessageConverter();
        return jsonMessageConverter;
    }

    @Autowired
    @Qualifier(value = "myRabbitTemplate")
    private RabbitTemplate myRabbitTemplate;

	/**
	 * 每5秒钟发送一个随机数字到队列
	 */
	// @Scheduled(cron = "0/5 * * * * ?")
	public void send() {
		try {
			Student stu = new Student();
			stu.setId(UUID.randomUUID().toString());
            stu.setName(UUID.randomUUID().toString());
			// 获取自定义的 MyCorrelationData对象
			ObjectMapper mapper = new ObjectMapper();
			MyCorrelationData myCorrelationData = new MyCorrelationData(UUID.randomUUID().toString(),
					mapper.writeValueAsString(stu));
			// object:发送的消息内容 correlationData:消息ID
			// 这里定义的 UTFMessagePostProcessor， 方便根据消息超时时间，确定是否处理到死信队列
            myRabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME, RabbitMQConfig.ROUTING_KEY_MSG, stu,
					// new UTFMessagePostProcessor(),
					myCorrelationData);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	@Override
	public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
		// 当消息发送出去找不到对应路由队列时，将会把消息退回
		// 如果有任何一个路由队列接收投递消息成功，则不会退回消息
		System.out.println("消息退回：" + new String(message.getBody()));
	}

	@Override
	public void confirm(org.springframework.amqp.rabbit.support.CorrelationData correlationData, boolean ack,
			String cause) {
		// ACK=true仅仅标示消息已被Broker接收到，并不表示已成功投放至消息队列中
		// ACK=false标示消息由于Broker处理错误，消息并未处理成功
		if (correlationData instanceof MyCorrelationData) {
			MyCorrelationData myCorrelationData = (MyCorrelationData) correlationData;
			System.out.println("消息送达确认结果：" + ack + ", 消息体:" + myCorrelationData.getMessage());
		}
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		// 必须设置消息送达确认的方式
        myRabbitTemplate.setConfirmCallback(this);
        myRabbitTemplate.setReturnCallback(this);
        myRabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
	}

}